/** @file    CommunicationManagerBundle.cpp
 *  @author  Alessandro Polo
 *  @version $Id: CommunicationManagerBundle.cpp 2832 2010-08-02 22:53:46Z alex $
 *  @brief
 * File containing methods for the wosh::services::CommunicationManagerBundle class.
 * The header for this class can be found in CommunicationManagerBundle.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2010, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/

 #include "CommunicationManagerBundle.h"

 #include "CommunicationManagerImpl.h"
 #include "CommunicationManagerParser.h"

 #include <core/ObjectAllocator.h>
 #include <core/MethodsCommon.h>
 #include <core/MessageProcessingCommon.h>
 #include <core/MessageFilter.h>
 #include <core/Fact.h>
 #include <core/List.h>
 #include <core/Table.h>
 #include <core/PersistenceManager.h>
 #include <core/FileSystem.h>
 #include <core/SystemError.h>

 #include <framework/communication/SessionIM.h>
 #include <framework/communication/TextMessage.h>

 #include <stdlib.h>
 #include <time.h>
 #include <algorithm>

using namespace std;
namespace wosh {
 using namespace communication;
 namespace services {

 WOSH_REGISTER(wosh::services::CommunicationManagerBundle, "wosh::interfaces::services::CommunicationManager", _CommunicationManager_VERSION, _static_CommunicationManagerBundle1 )
 WOSH_REGISTER(wosh::services::CommunicationManagerBundle, "wosh::BundleGeneric", _CommunicationManager_VERSION, _static_CommunicationManagerBundle2 )

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

CommunicationManagerBundle::CommunicationManagerBundle()
	: BundleGeneric() {
	BundleGeneric::setName( _CommunicationManager_NAME, false );
	Object::getURI().setKernelLocal();
	Object::getURI().clearPathName();
	Object::getURI().setName( "Communication" );
	Object::getURI().registerGlobal(); // so BundleGeneric won't overwrite my uri
	Properties.update( _KEY_Version, _CommunicationManager_VERSION );

	Requirements.add<wosh::communication::SessionIM>();
	Requirements.add<wosh::communication::NotificationFlow>();
	Requirements.add<wosh::communication::TextMessage>();

//Log.setLevel(LOG_VERBOSE);

	Log(LOG_DEBUG, " Configuring CommunicationManager workers.." );
	this->commWorker = new CommunicationManagerImpl(*this);
	this->commWorker->setListener(this);
	this->commWorker->setFlowTimeOut( 120 );
	this->commWorker->setFlowRetryTime( 20 );

	this->commParser = new CommunicationManagerParser(*this, this->commWorker->getCommunicators(), this->commWorker->getInterpreters());
	this->commParser->setThreadListener(NULL);
	this->commParser->setTimeOut( 0 );

	Log(LOG_DEBUG, " Setting default properties and permissions.." );
	Properties.set( _CommunicationManager_KEY_ServiceTimeOut, this->commParser->getTimeOut(), Permission( Permission::RX, Permission::Read, Permission::Read ) );
	Properties.set( _CommunicationManager_KEY_FlowTimeOut, this->commWorker->getFlowTimeOut(), Permission( Permission::RX, Permission::Read, Permission::Read ) );
	Properties.set( _CommunicationManager_KEY_FlowRetryTime, this->commWorker->getFlowRetryTime(), Permission( Permission::RX, Permission::Read, Permission::Read ) );
	Properties.monitor(_CommunicationManager_KEY_ServiceTimeOut);
	Properties.monitor(_CommunicationManager_KEY_FlowTimeOut);
	Properties.monitor(_CommunicationManager_KEY_FlowRetryTime);

//	Properties.set( "SaveOnExit", true, Permission( Permission::RX, Permission::RX, Permission::Read ) );
//save archived
//save on exit

	Log(LOG_DEBUG, " Registering methods.." );
	MethodList* mmListCommunicators = Methods.create<MethodList>( _CommunicationManager_METHOD_communicators, "List Communicators", Permission(Permission::RX) );
	mmListCommunicators->setDataFieldExtractor( DataFieldExtractor<Communicator>::createFor< DataAdapterMapT<string, Communicator> >(this->commWorker->getCommunicators(), "Communicators") );

	MethodRetrieve* mmRetrieveCommunicators = Methods.create<MethodRetrieve>( _METHOD_Retrieve, "Retrieve Communicator(s)", Permission::R_XR_XR__() );
	mmRetrieveCommunicators->setItemExtractor( ObjectExtractor<Communicator, const std::string&>::createFor< DataAdapterMapT<string, Communicator> >(this->commWorker->getCommunicators(), &Communicator::getName ) );

	MethodList* mmListInterpreters = Methods.create<MethodList>( _CommunicationManager_METHOD_interpreters, "List Interpreters", Permission(Permission::RX) );
	mmListInterpreters->setDataFieldExtractor( DataFieldExtractor<Interpreter>::createFor< DataAdapterMapT<string, Interpreter> >(this->commWorker->getInterpreters(), "Interpreters") );

	MethodRetrieve* mmRetrieveInterpreters = Methods.create<MethodRetrieve>( _METHOD_Retrieve, "Retrieve Interpreter(s)", Permission::R_XR_XR__() );
	mmRetrieveInterpreters->setItemExtractor( ObjectExtractor<Interpreter, const std::string&>::createFor< DataAdapterMapT<string, Interpreter> >(this->commWorker->getInterpreters(), &Interpreter::getName ) );

	MethodRequest* mmListMessages = Methods.create<MethodRequest>( _CommunicationManager_METHOD_messages, "List messages", Permission(Permission::RX) );
	mmListMessages->setMethod( this, (MethodRequestPtr)&CommunicationManagerBundle::mmDoListMessages );

	MethodRequest* mmListMessagesAr = Methods.create<MethodRequest>( _CommunicationManager_METHOD_messages_archived, "List archived messages", Permission(Permission::RX) );
	mmListMessagesAr->setMethod( this, (MethodRequestPtr)&CommunicationManagerBundle::mmDoListMessagesAr );

	MethodMessageResponse* mmListMessagesMy = Methods.create<MethodMessageResponse>( _CommunicationManager_METHOD_messages_my, "List my messages", Permission(Permission::RX) );
	mmListMessagesMy->setMethod( this, (MethodMessageResponsePtr)&CommunicationManagerBundle::mmDoListMessagesMy );

	MethodMessageResponse* mmMark = Methods.create<MethodMessageResponse>( _CommunicationManager_METHOD_mark, "Mark a message as..", Permission(Permission::RX) );
	mmMark->setMethod( this, (MethodMessageResponsePtr)&CommunicationManagerBundle::mmDoMark );

	MethodWRESULT* mmSaveMessages = Methods.create<MethodWRESULT>( _CommunicationManager_METHOD_messages_dump, "save messages to xml file", Permission::R_XR__R__() );
	mmSaveMessages->setMethod( this, (MethodWRESULTPtr)&CommunicationManagerBundle::mmDoSaveMessages );

	MethodMessageResponse* mmSendTell = Methods.create<MethodMessageResponse>( _CommunicatorService_METHOD_sendmessage, "send a message..", Permission(Permission::RX) );
	mmSendTell->setMethod( this, (MethodMessageResponsePtr)&CommunicationManagerBundle::mmDoTell );

	MethodRequest* mmFakeCommunicatorEvent = Methods.create<MethodRequest>( _CommunicationManager_METHOD_fake_event, "fake Communicator event", Permission::R_XR__R__() );
	mmFakeCommunicatorEvent->setMethod( this, (MethodRequestPtr)&CommunicationManagerBundle::mmDoFakeCommunicatorEvent );

	MethodWRESULT* mmUpdateCommunicators = Methods.create<MethodWRESULT>( _CommunicationManager_METHOD_update_comm, "force update Communicator", Permission::R_XR__R__() );
	mmUpdateCommunicators->setMethod( this, (MethodWRESULTPtr)&CommunicationManagerBundle::updateCommunicators );

	MethodWRESULT* mmUpdateInterpreters = Methods.create<MethodWRESULT>( _CommunicationManager_METHOD_update_inter, "force update Interpreters", Permission::R_XR__R__() );
	mmUpdateInterpreters->setMethod( this, (MethodWRESULTPtr)&CommunicationManagerBundle::updateInterpreters );

	setBundleState(Bundle::STATE_CREATED, false);
 }


CommunicationManagerBundle::~CommunicationManagerBundle() {
	Log(LOG_INFO, " Destroying.." );
	if ( isBundleRunning() ) {
		Log(LOG_WARNING, "~CommunicationManagerBundle() : Destroying while RUNNING! Trying to stop.." );
		bundleStop();
	 }
	delete this->commParser; this->commParser = NULL;
	delete this->commWorker; this->commWorker = NULL;
	Log(LOG_VERBOSE, ":~CommunicationManagerBundle() : Destroyed." );
 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////// BUNDLE CONTROL

WRESULT CommunicationManagerBundle::bundleStart() {
	if ( !BundleGeneric::bundleValidate_StartStop(Bundle::STATE_STARTING) ) return WRET_ERR_WRONG_STATE;

	WRESULT ret = WRET_OK;
	ret += this->commParser->startThread( 15000 );
	if ( WFAILED(ret) ) {
		SystemError::raise(this, Log, SystemError::CRITICAL, 0, "bundleStart", "FAILED waiting RUNNING state of commParser.." );
		return WRET_ERR_INTERNAL;
	 }
	ret += BundleGeneric::start_SynchThread( this->commWorker );
	// BUNDLE-STATE (STARTED) will be updated async by WORKER, through call: thread_event()

	updateCommunicators();
	updateInterpreters();

	return ret;
 }

WRESULT CommunicationManagerBundle::bundleStop() {
	if ( !BundleGeneric::bundleValidate_StartStop(Bundle::STATE_STOPPING) ) return WRET_ERR_WRONG_STATE;

	WRESULT ret = WRET_OK;
	ret += this->commParser->quitThread( 15000 );
	if ( WFAILED(ret) ) {
		SystemError::raise(this, Log, SystemError::CRITICAL, 0, "bundleStop", "FAILED waiting STOPPED state of commParser.." );
		return WRET_ERR_INTERNAL;
	 }
	ret += BundleGeneric::stop_SynchThread( this->commWorker );
	// BUNDLE-STATE (STOPPED) will be updated async by WORKER, through call: thread_event()
	return ret;
 }

////////////////////////////////////////////////////////////////////////////////////////////// BUNDLE CONTROL
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////// PROCESS BUS MESSAGES

void CommunicationManagerBundle::busMessage( const Message& message, const Bus* ) {
	if ( message.isEmpty() ) return;
	// core business:
	if ( message.getContent()->isNotification() ) {
		// monitor (incoming) notifications (for kernel)
		// monitor (outgoing) notifications (for users)
		if ( message.getSource() == this->getURI() ) return; // sent by myself?
		this->commWorker->evalMessage( message );
		return;
	 }

	if ( !MessageFilter::isMessageTarget(message, this) )
		return;

	if ( message.getContent()->isRequest() ) {
		Message* reply = MPC_busMessageRequest( message, this->Methods, this->Log );
		if ( reply == NULL )
			return;
		signMessage(reply);
		BusCore.postMessage(reply);
		return;
	 }

	// management
	// monitor new/removed hosts (to update remote Communicators)
	// monitor bundle state-changed/install/removal
	// request/update communicators list
	// retrieve communicators info

	this->commParser->evalMessage( message );

	if ( message.getContent()->isResponse() ) {
		// maybe is response of a SENDING
		this->commWorker->evalMessage( message );
	 }
 }

//////////////////////////////////////////////////////////////////////////////////////// PROCESS BUS MESSAGES
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////// PROPERTY EVENT

bool CommunicationManagerBundle::updatingProperty( bool& do_update, const Variant& value_proposed, Property& property_current, const PropertiesProvider* source )
 {
	if ( property_current.getKey() == _CommunicationManager_KEY_ServiceTimeOut ) {
		this->commParser->setTimeOut( value_proposed.toLong( this->commParser->getTimeOut() ) );
		return true;
	 }
	else if ( property_current.getKey() == _CommunicationManager_KEY_FlowTimeOut ) {
		this->commWorker->setFlowTimeOut( value_proposed.toLong( this->commWorker->getFlowTimeOut() ) );
		return true;
	 }
	else if ( property_current.getKey() == _CommunicationManager_KEY_FlowRetryTime ) {
		this->commWorker->setFlowRetryTime( value_proposed.toLong( this->commWorker->getFlowRetryTime() ) );
		return true;
	 }
	else // maybe is a (common) BundleGeneric property, let base implementation process it
		return BundleGeneric::updatingProperty(do_update, value_proposed, property_current, source);
 }

////////////////////////////////////////////////////////////////////////////////////////////// PROPERTY EVENT
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT CommunicationManagerBundle::mmDoSaveMessages()
 {
	FilePath xmlFile;
	xmlFile.set("$DATABASE/notifications.xml");
	Log(LOG_VERBOSE, ":mmDoSaveMessages() : Saving to %s", xmlFile.getPathLocal().c_str() );
	this->commWorker->getMessages().transactionBeginRead();
	WRESULT ret = PersistenceManager::saveObjects( this->commWorker->getMessages().getMap(), "XML", xmlFile.getPathLocal() );
	this->commWorker->getMessages().transactionEnd();
	if ( WFAILED(ret) ) {
		SystemError::raise(this, Log, SystemError::CRITICAL, 0, "mmDoSaveMessages", "FAILED#%d saving Database %s..", ret,"-" );
	 }

	xmlFile.set("$DATABASE/notifications.archived.xml");
	Log(LOG_VERBOSE, ":mmDoSaveMessages() : Saving to %s", xmlFile.getPathLocal().c_str() );
	this->commWorker->getMessagesArchived().transactionBeginRead();
	 ret = PersistenceManager::saveObjects( this->commWorker->getMessagesArchived().getMap(), "XML", xmlFile.getPathLocal() );
	this->commWorker->getMessagesArchived().transactionEnd();
	return ret;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////// METHODS

WRESULT CommunicationManagerBundle::updateCommunicators()	{ return this->commParser->updateCommunicators(); }
WRESULT CommunicationManagerBundle::updateInterpreters()	{ return this->commParser->updateInterpreters(); }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

Response* CommunicationManagerBundle::mmDoTell( const Message* requestMessage )
 {
	if ( requestMessage == NULL || requestMessage->isEmpty() ) return NULL;
	if ( !requestMessage->getContent()->isRequest() ) return NULL;
	const Request* request = requestMessage->getContent()->asRequest();
	Variant dUserName;
	Variant dMessage;
	request->extractArguments(dUserName, dMessage);
	if ( !dUserName.isStringNotEmpty() || !dMessage.isStringNotEmpty() )
		return request->replyResponse(WRET_ERR_PARAM, "Invalid arguments [username, message]");
	std::string sender = "";
	std::string text_message = dMessage.toString();
	if ( requestMessage->getSecurityToken() != NULL )
		sender = requestMessage->getSecurityToken()->getUsername();
	if ( sender != dUserName.asString() && sender != "wosh ")
		text_message += " [by " + sender + "]";
	TextMessage* notif = new TextMessage();
	notif->setSender_User(sender);
	notif->setPlain( text_message );

	this->commWorker->getMessages().transactionBeginWrite();
	NotificationFlow* nEx = this->commWorker->evalNotification_(notif, requestMessage);
	bool sent = this->commWorker->sendNotification_( nEx );
	this->commWorker->getMessages().transactionEnd();

	if ( sent )
		return request->replyResponse(WRET_OK, "Sent!" );
	else
		return request->replyResponse(WRET_ERR_INTERNAL, "Internal Error!");
 }

Response* CommunicationManagerBundle::mmDoFakeCommunicatorEvent( const Request* request )
 {
	if ( request == NULL ) return NULL;
	Variant dDeviceName;
	Variant dEventName;
	Variant dEventData;
	request->extractArguments(dDeviceName, dEventName, dEventData);
	if ( !dDeviceName.isStringNotEmpty() || !dEventName.isStringNotEmpty() )
		return request->replyResponse(WRET_ERR_PARAM, "Invalid arguments [username, location]");
	Log(LOG_VERBOSE, "mmDoFakeCommunicatorEvent(%s) Fact: %s [%s]", dDeviceName.asString().c_str(), dEventName.asString().c_str(), dEventData.toString().c_str() );

	Message* msg_event = new Message();
	msg_event->setSource( this );
	msg_event->getSource().setParent(this->getURI());
	msg_event->getSource().setName( dDeviceName.asString() );
	msg_event->setContent( new Fact(dEventName.asString(), new Variant(dEventData)) );
	msg_event->setDestinationBroadcast();
	BusCore.postMessage(msg_event);
	return request->replyResponse(WRET_OK, "Communicator-Event (message) sent!" );
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

Response* CommunicationManagerBundle::mmDoListMessages( const Request* request )
 {
	if ( request == NULL ) return NULL;
	string option = request->getArgument().toString();
	Log(LOG_VERBOSE, ":mmDoListMessages(%s)", option.c_str() );

	if ( option == "s" || option == "short" ) {
		this->commWorker->getMessages().transactionBeginRead();
		List* objs =  new List(this->commWorker->getMessages().size());
		MapT<long, NotificationFlow*>::ConstIterator it;
		MapT<long, NotificationFlow*>::ConstIterator it_end = this->commWorker->getMessages().end();
		for ( it=this->commWorker->getMessages().begin(); it!=it_end; ++it ) {
			if ( it->second == NULL || !it->second->hasOriginal() ) continue;
			objs->push_back<Variant>( it->second->getOriginal()->getID() );
		 }
		this->commWorker->getMessages().transactionEnd();
		return request->replyResponse(WRET_OK, objs);
	 }
	else {
		this->commWorker->getMessages().transactionBeginRead();
		Table* tbl = new Table( this->commWorker->getMessages().size(), 11, true );
		tbl->setTableName("Messages");
		tbl->setHeaderHorizontal("F-ID", 0);
		tbl->setHeaderHorizontal("N-ID", 1);
		tbl->setHeaderHorizontal("TimeStamp", 2);
		tbl->setHeaderHorizontal("Sender", 3);
		tbl->setHeaderHorizontal("Recipent", 4);
		tbl->setHeaderHorizontal("Received By", 5);
		tbl->setHeaderHorizontal("Sent with", 6);
		tbl->setHeaderHorizontal("Sent At", 7);
		tbl->setHeaderHorizontal("Retries", 8);
		tbl->setHeaderHorizontal("Errors", 9);
		tbl->setHeaderHorizontal("Type", 10);
		unsigned int iRow = 0;
		MapT<long, NotificationFlow*>::ConstIterator it;
		MapT<long, NotificationFlow*>::ConstIterator it_end = this->commWorker->getMessages().end();
		for ( it=this->commWorker->getMessages().begin(); it!=it_end; ++it ) {
			if ( it->second == NULL || !it->second->hasOriginal() ) continue;
			tbl->set<Variant>( it->second->getID(), iRow, 0);
			tbl->set<Variant>( it->second->getOriginal()->getID(), iRow, 1);
			tbl->set( new Variant(it->second->getOriginal()->getTimeStamp(), Variant::DATETIME), iRow, 2);
			tbl->set<Variant>( it->second->getOriginal()->getSender_User(), iRow, 3);
			tbl->set<Variant>( it->second->getOriginal()->getRecipent_User(), iRow, 4);
			tbl->set( it->second->getSourceCommunicator().clone(), iRow, 5);
			tbl->set( it->second->getSenderCommunicator().clone(), iRow, 6);
			if ( it->second->getOriginal()->isSent() )
					tbl->set( new Variant(it->second->getOriginal()->getSentTimeStamp(), Variant::DATETIME), iRow, 7);
			else	tbl->set( new Variant(false), iRow, 7);
			tbl->set<Variant>( it->second->getSentCount(), iRow, 8);
			tbl->set<Variant>( it->second->getErrorsCount(), iRow, 9);
			tbl->set<Variant>( (int)it->second->getOriginal()->getNotificationType(), iRow, 10);
			++iRow;
		 }
		this->commWorker->getMessages().transactionEnd();
		return request->replyResponse(WRET_OK, tbl);
	 }
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

Response* CommunicationManagerBundle::mmDoListMessagesAr( const Request* request )
 {
	if ( request == NULL ) return NULL;
	string option = request->getArgument().toString();
	Log(LOG_VERBOSE, ":mmDoListMessagesAr(%s)", option.c_str() );

	if ( option == "s" || option == "short" ) {
		this->commWorker->getMessagesArchived().transactionBeginRead();
		List* objs =  new List(this->commWorker->getMessagesArchived().size());
		MapT<long, NotificationFlow*>::ConstIterator it;
		MapT<long, NotificationFlow*>::ConstIterator it_end = this->commWorker->getMessagesArchived().end();
		for ( it=this->commWorker->getMessagesArchived().begin(); it!=it_end; ++it ) {
			if ( it->second == NULL || !it->second->hasOriginal() ) continue;
			objs->push_back<Variant>( it->second->getOriginal()->getID() );
		 }
		this->commWorker->getMessagesArchived().transactionEnd();
		return request->replyResponse(WRET_OK, objs);
	 }
	else {
		this->commWorker->getMessagesArchived().transactionBeginRead();
		Table* tbl = new Table( this->commWorker->getMessagesArchived().size(), 12, true );
		tbl->setTableName("Messages");
		tbl->setHeaderHorizontal("F-ID", 0);
		tbl->setHeaderHorizontal("N-ID", 1);
		tbl->setHeaderHorizontal("TimeStamp", 2);
		tbl->setHeaderHorizontal("Sender", 3);
		tbl->setHeaderHorizontal("Recipent", 4);
		tbl->setHeaderHorizontal("Sent At", 5);
		tbl->setHeaderHorizontal("Read At", 6);
		tbl->setHeaderHorizontal("Received By", 7);
		tbl->setHeaderHorizontal("Sent with", 8);
		tbl->setHeaderHorizontal("Retries", 9);
		tbl->setHeaderHorizontal("Errors", 10);
		tbl->setHeaderHorizontal("Type", 11);
		unsigned int iRow = 0;
		MapT<long, NotificationFlow*>::ConstIterator it;
		MapT<long, NotificationFlow*>::ConstIterator it_end = this->commWorker->getMessagesArchived().end();
		for ( it=this->commWorker->getMessagesArchived().begin(); it!=it_end; ++it ) {
			if ( it->second == NULL || !it->second->hasOriginal() ) continue;
			tbl->set<Variant>( it->second->getID(), iRow, 0);
			tbl->set<Variant>( it->second->getOriginal()->getID(), iRow, 1);
			tbl->set( new Variant(it->second->getOriginal()->getTimeStamp(), Variant::DATETIME), iRow, 2);
			tbl->set<Variant>( it->second->getOriginal()->getSender_User(), iRow, 3);
			tbl->set<Variant>( it->second->getOriginal()->getRecipent_User(), iRow, 4);
			if ( it->second->getOriginal()->isSent() )
					tbl->set( new Variant(it->second->getOriginal()->getSentTimeStamp(), Variant::DATETIME), iRow, 5);
			else	tbl->set( new Variant(false), iRow, 5);
			if ( it->second->getOriginal()->isRead() )
					tbl->set( new Variant(it->second->getOriginal()->getReadTimeStamp(), Variant::DATETIME), iRow, 6);
			else	tbl->set( new Variant(false), iRow, 6);
			tbl->set( it->second->getSourceCommunicator().clone(), iRow, 7);
			tbl->set( it->second->getSenderCommunicator().clone(), iRow, 8);
			tbl->set<Variant>( it->second->getSentCount(), iRow, 9);
			tbl->set<Variant>( it->second->getErrorsCount(), iRow, 10);
			tbl->set<Variant>( (int)it->second->getOriginal()->getNotificationType(), iRow, 11);
			++iRow;
		 }
		this->commWorker->getMessagesArchived().transactionEnd();
		return request->replyResponse(WRET_OK, tbl);
	 }
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

Response* CommunicationManagerBundle::mmDoListMessagesMy( const Message* requestMessage )
 {
	if ( requestMessage == NULL || requestMessage->isEmpty() ) return NULL;
	if ( !requestMessage->getContent()->isRequest() ) return NULL;
	const Request* request = requestMessage->getContent()->asRequest();
	if ( request == NULL ) return NULL;
	if ( !requestMessage->hasSecurityToken() ) return NULL;
	string user = requestMessage->getSecurityToken()->getUsername();
	Log(LOG_VERBOSE, ":mmDoListMessagesMy(%s)", user.c_str() );

	this->commWorker->getMessages().transactionBeginRead();
	unsigned int iRow = 0;
	MapT<long, NotificationFlow*>::ConstIterator it;
	MapT<long, NotificationFlow*>::ConstIterator it_end = this->commWorker->getMessages().end();
	for ( it=this->commWorker->getMessages().begin(); it!=it_end; ++it ) {
		if ( it->second == NULL || !it->second->hasOriginal() ) continue;
		if ( it->second->getOriginal()->getRecipent_User() != user ) continue;
		++iRow;
	 }
	Table* tbl = new Table( iRow, 3, true, true );
	tbl->setTableName("Messages of " + user);
	tbl->setHeaderHorizontal("TimeStamp", 0);
	tbl->setHeaderHorizontal("Sender", 1);
	tbl->setHeaderHorizontal("Plain", 2);
	iRow = 0;
	for ( it=this->commWorker->getMessages().begin(); it!=it_end; ++it ) {
		if ( it->second == NULL || !it->second->hasOriginal() ) continue;
		if ( it->second->getOriginal()->getRecipent_User() != user ) continue;
		tbl->setHeaderVertical( it->second->getOriginal()->getID(), iRow );
		tbl->set( new Variant(it->second->getOriginal()->getTimeStamp(), Variant::DATETIME), iRow, 0);
		tbl->set<Variant>( it->second->getOriginal()->getSender_User(), iRow, 1);
		tbl->set<Variant>( it->second->getOriginal()->getPlain(), iRow, 2);
		++iRow;
	 }

	this->commWorker->getMessages().transactionEnd();
	return request->replyResponse(WRET_OK, tbl);
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

Response* CommunicationManagerBundle::mmDoMark( const Message* requestMessage )
 {
	if ( requestMessage == NULL || requestMessage->isEmpty() ) return NULL;
	if ( !requestMessage->getContent()->isRequest() ) return NULL;
	const Request* request = requestMessage->getContent()->asRequest();
	if ( request == NULL ) return NULL;
	if ( !requestMessage->hasSecurityToken() ) return NULL;
	string user = requestMessage->getSecurityToken()->getUsername();
	Variant vId;
	Variant vFlag;
	request->extractArguments(vId, vFlag);
	if ( vId.toLong(0) == 0 || !vFlag.isStringNotEmpty() )
		return request->replyResponse(WRET_ERR_PARAM, "invalid arguments [id, flags]");
	Log(LOG_VERBOSE, ":mmDoMark(%ld, %s) user: %s", vId.toLong(0), vFlag.asString().c_str(), user.c_str() );

	this->commWorker->getMessages().transactionBeginRead();
	MapT<long, NotificationFlow*>::ConstIterator it;
	MapT<long, NotificationFlow*>::ConstIterator it_end = this->commWorker->getMessages().end();
	for ( it=this->commWorker->getMessages().begin(); it!=it_end; ++it ) {
		if ( it->second == NULL || !it->second->hasOriginal() ) continue;
		if ( it->second->getOriginal()->getRecipent_User() != user ) continue;
		if ( it->second->getOriginal()->getID() != vId.toLong(0) ) continue;
		this->commWorker->archiveNotification_(it->second);
		this->commWorker->getMessages().transactionEnd();
		return request->replyResponse(WRET_OK, "Archieved" );
	 }
	this->commWorker->getMessages().transactionEnd();
	return request->replyResponse(WRET_ERR_PARAM, "Message not found");
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
 }; // namespace services
}; // namespace wosh
